package com.yujiahao.bigdata.teacher

import org.apache.spark.{SparkConf, SparkContext}

object Spark02_Req_HotCategoryTop10SessionAnalysis {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("HotCategoryTop10SessionAnalysis").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // TODO 需求一
    val rdd = sc.textFile("data/user_visit_action.txt")
    rdd.cache()
    val top10 = rdd.flatMap(
      data => {
        val datas = data.split("_")
        if (datas(6) != "-1") {
          // 数据为点击的场合
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          // 数据为下单的场合
          val ids = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
        } else if (datas(10) != "null") {
          // 数据为支付的场合
          val ids = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
        } else {
          Nil
        }
      }
    ).reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    ).sortBy(_._2, false).take(10)
    val top10Ids: Array[String] = top10.map(_._1)

    // TODO 需求二 *****************************************************

    // 1. 将数据进行筛选过滤，只保留Top10热门品类的数据
    // 1.1 将数据转换为样例类
    val actionRDD = rdd.map(
      line => {
        val datas = line.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    // 1.2 将数据进行过滤

    val filterRDD = actionRDD.filter(
      action => {
        if (action.click_category_id != -1) {
          // 点击数据
          top10Ids.contains(action.click_category_id.toString)
        } else {
          false
        }
      }
    )

    // 2. 将数据进行结构的转换
    //    line => ( (品类，session), 1 )
    // 3. 将转换结构后的数据进行统计
    //     ( (品类，session), 1 ) =>  ( (品类，session), sum )
    val reduceRDD = filterRDD.map(
      action => {
        ((action.click_category_id, action.session_id), 1)
      }
    ).reduceByKey(_ + _)

    // 4. 将统计结果进行结构的转换
    //     ( (品类，session), sum ) =>  ( 品类, (session, sum) )
    // 5. 将转换结构后的数据按照品类进行分组
    //     ( 品类, (session, sum) ) => ( 品类, Iter[(session, sum),(session1, sum1),(session2, sum2)] )

    val groupRDD = reduceRDD.map {
      case ((cid, sessionid), sum) => {
        (cid, (sessionid, sum))
      }
    }.groupByKey()

    // 6. 将分组后的数据进行排序（降序），取前10名
    val result = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )

    // 7. 将结果采集后打印在控制台
    result.collect().foreach(println)

    sc.stop()

  }

  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, //用户的ID
                              session_id: String, //Session的ID
                              page_id: Long, //某个页面的ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, //某一个商品品类的ID
                              click_product_id: Long, //某一个商品的ID
                              order_category_ids: String, //一次订单中所有品类的ID集合
                              order_product_ids: String, //一次订单中所有商品的ID集合
                              pay_category_ids: String, //一次支付中所有品类的ID集合
                              pay_product_ids: String, //一次支付中所有商品的ID集合
                              city_id: Long //城市 id
                            )
}
